import org.apache.hadoop.hbase.{HBaseConfiguration, TableName}
import org.apache.hadoop.hbase.client._
import org.apache.hadoop.hbase.util.Bytes
import scala.collection.JavaConverters._
import org.apache.log4j.{Level, Logger}

object Main {
  def main(args: Array[String]): Unit = {

    // 启用HBase客户端调试日志
    Logger.getLogger("org.apache.hadoop.hbase").setLevel(Level.DEBUG)
    Logger.getLogger("org.apache.zookeeper").setLevel(Level.DEBUG)
    System.setProperty("hadoop.security.authentication", "simple")

    // 1. 创建配置和连接
    val conf = HBaseConfiguration.create()
    conf.set("hbase.zookeeper.quorum", "192.168.1.208:7077,192.168.1.208:7078,192.168.1.208:7079")
//    conf.set("hbase.zookeeper.property.clientPort", "7077")
//    conf.addResource("D:\\github\\spark-scala-tutorial\\HBaseClientRead\\hadoop1_hbase.conf")
//    conf.addResource("D:\\github\\spark-scala-tutorial\\HBaseClientRead\\hadoop2_hbase.conf")
//    conf.addResource("D:\\github\\spark-scala-tutorial\\HBaseClientRead\\hadoop3_hbase.conf")

    var connection: Connection = null
    try {
      connection = ConnectionFactory.createConnection(conf)

      // 2. 获取表引用
      val table = connection.getTable(TableName.valueOf("mapdata"))

      // 3. 单行读取
      val get = new Get(Bytes.toBytes("0000721a00002abc"))
        .addColumn(Bytes.toBytes("route"), Bytes.toBytes("accessible"))
      val result = table.get(get)
      println(s"User name are: ${Bytes.toString(result.getValue(Bytes.toBytes("route"), Bytes.toBytes("accessible")))}")

      // 4. 范围扫描
      val scan = new Scan()
        .withStartRow(Bytes.toBytes("0000000000000000"))
        .withStopRow(Bytes.toBytes("FFFFFFFFFFFFFFFF"))
        .addColumn(Bytes.toBytes("route"), Bytes.toBytes("accessible"))
        .setLimit(1)
        .setCaching(1)

      val scanner = table.getScanner(scan)
      try {
        scanner.asScala.take(10).foreach { result =>
          println(s"${Bytes.toString(result.getRow)}: ${Bytes.toString(result.getValue(Bytes.toBytes("route"), Bytes.toBytes("accessible")))}")
        }
      } finally {
        scanner.close()
      }

    } finally {
      if (connection != null) connection.close()
    }
  }
}